package com.alibaba.citrus.cr.platform.enhance.facade.sls;

import java.util.ArrayList;
import java.util.List;

import javax.annotation.Resource;

import com.alibaba.citrus.cr.platform.enhance.repository.OperateLogWriteRepository;
import com.alibaba.cz.base.tool.log.Log;
import com.alibaba.fastjson.JSON;

import com.aliyun.openservices.log.common.FastLog;
import com.aliyun.openservices.log.common.FastLogContent;
import com.aliyun.openservices.log.common.FastLogGroup;
import com.aliyun.openservices.log.common.LogGroupData;
import com.aliyun.openservices.loghub.client.ILogHubCheckPointTracker;
import com.aliyun.openservices.loghub.client.exceptions.LogHubCheckPointException;
import com.aliyun.openservices.loghub.client.interfaces.ILogHubProcessor;
import com.epoch.app.crplatformenhance.domain.operatelog.model.OperateLog;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

/**
 * @author: qiaojiange
 * @create: 2021/11/09
 * @Description 日志处理类
 */
@Component
@Scope(scopeName = "prototype")
public class LogHubProcessor implements ILogHubProcessor {
    /**
     * 普通日志
     */
    protected final Log log = Log.getLogger(LogHubProcessor.class);

    /**
     * 记录上次持久化Checkpoint的时间。
     */
    private long lastCheckTime = 0;

    @Resource
    private DiffProcessor diffProcessor;

    @Resource
    private OperateLogWriteRepository writeRepository;

    private int shardId;

    @Override
    public void initialize(int shardId) {
        this.shardId = shardId;
    }

    @Override
    public String process(List<LogGroupData> logGroups, ILogHubCheckPointTracker checkPointTracker) {
        try {
            List<OperateLog> logModelList = new ArrayList<>();
            // 将获取到的数据打印出来。
            for (LogGroupData logGroup : logGroups) {
                FastLogGroup fastLogGroup = logGroup.GetFastLogGroup();
                for (int i = 0; i < fastLogGroup.getLogsCount(); i++) {
                    FastLog fastLog = fastLogGroup.getLogs(i);
                    OperateLog operateLog = null;
                    for (int j = 0; j < fastLog.getContentsCount(); j++) {
                        FastLogContent content = fastLog.getContents(j);
                        String key = content.getKey();
                        String value = content.getValue();
                        if ("message".equals(key)) {
                            operateLog = JSON.parseObject(value, OperateLog.class);
                            diffProcessor.diff(operateLog);
                            logModelList.add(operateLog);
                        }
                    }
                }
            }

            if (CollectionUtils.isNotEmpty(logModelList)) {
                writeRepository.batchSaveOperateLog(logModelList);
            }


            long curTime = System.currentTimeMillis();
            // 每隔30秒，写一次Checkpoint到服务端，如果30秒内发生Worker crash，新启动的Worker会从上一个Checkpoint获取消费数据，可能存在少量的重复数据。
            if (curTime - lastCheckTime > 30 * 1000) {
                try {
                    //参数为true表示立即将Checkpoint更新到服务端，为false会将Checkpoint缓存在本地，默认隔60s会将Checkpoint更新到服务端。
                    checkPointTracker.saveCheckPoint(true);
                } catch (LogHubCheckPointException e) {
                    e.printStackTrace();
                }
                lastCheckTime = curTime;
            }
        } catch (Throwable e) {
            log.error("日志消费失败", e);
        }
        return null;
    }

    // 当Worker退出时，会调用该函数，您可以在此处执行清理工作。
    @Override
    public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
        //将Checkpoint保存到服务端。
        try {
            checkPointTracker.saveCheckPoint(true);
        } catch (LogHubCheckPointException e) {
            e.printStackTrace();
            log.error("关闭出异常了，需要处理!", e);
        }
    }
}
